package com.study.chapter12;

import com.study.entity.LoginEvent;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/28 17:20
 */
public class LoginFailDetectProExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1.获取登录数据流
        SingleOutputStreamOperator<LoginEvent> loginEventStream = env.fromElements(
                new LoginEvent("user_1", "192.168.0.1", "fail", 2000L),
                new LoginEvent("user_1", "192.168.0.2", "fail", 3000L),
                new LoginEvent("user_2", "192.168.1.29", "fail", 4000L),
                new LoginEvent("user_1", "171.56.23.10", "fail", 5000L),
                new LoginEvent("user_2", "192.168.1.29", "success", 6000L),
                new LoginEvent("user_2", "192.168.1.29", "fail", 7000L),
                new LoginEvent("user_2", "192.168.1.29", "fail", 8000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<LoginEvent>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner((event, l) -> event.timestamp)
        );

        // 2.定义模式，连续三次登录失败
        Pattern<LoginEvent, LoginEvent> eventPattern = Pattern.<LoginEvent>begin("fail")
                .where(new SimpleCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent value) throws Exception {
                        return StringUtils.equals(value.eventType, "fail");
                    }
                }).times(3).consecutive(); //指定严格近邻的三次登录失败

        // 3.将模式应用到数据流上，检测复杂事件
        PatternStream<LoginEvent> patternStream = CEP.pattern(loginEventStream.keyBy(event -> event.userId), eventPattern);

        //4. 将检测到的复杂事件提取出来，进行处理得到报警信息输出
        SingleOutputStreamOperator<String> warnStream = patternStream.process(new PatternProcessFunction<LoginEvent,String>(){

            @Override
            public void processMatch(Map<String, List<LoginEvent>> pattern, Context ctx, Collector<String> out) throws Exception {
                //提取复杂事件中的三次登录失败事件
                List<LoginEvent> failList = pattern.get("fail");
                LoginEvent first = failList.get(0);
                LoginEvent second = failList.get(1);
                LoginEvent third = failList.get(2);
               out.collect( first.userId + " 连续三次登录失败！登录时间："
                       + first.timestamp + ","
                       + second.timestamp + ","
                       + third.timestamp);
            }
        });

        warnStream.print("提示：");

        env.execute();
    }
}
